package logic

import (
	"encoding/json"
	"strconv"
	"time"

	"github.com/Shopify/sarama"
	"gitee.com/leicc/go-core"
	"gitee.com/leicc/go-orm/log"
	"gitee.com/leicc/go-app-core/logics"
)

const (
	HEARTINTVAL = 60
	IMMQKEY = "im@queue"
	IMGROUP = "IMPushGroup"
)

//0-私聊 1-频道群聊 2-世界频道 3-商品私聊
type QueueMessageSt struct {
	Type 	int16 	`json:"type"`
	UserId 	int64 	`json:"userid"`
	ToId	int64 	`json:"touserid"`
	Message []byte 	`json:"message"`
}
//定义消息通知的处理回调函数
type MQueueNotifyHandle func(int64, *QueueMessageSt)

type MQueueSt struct {
	mq     *core.KafkaMQ
	config *Config
	handle MQueueNotifyHandle
}

var GMQueue *MQueueSt = nil
func init()  {
	GMQueue = &MQueueSt{mq: nil, config: nil, handle: nil}
}

//释放删除redis当中的消息队列
func (c *MQueueSt) Release()  {
	//释放 消息队列数据
	if c.mq != nil {
		c.mq.Release()
	}
	if c.config != nil && len(c.config.MQueue) > 0 {
		logics.CRedis.HDel(IMMQKEY, c.config.MQueue)
	}
}

//初始化创建消息队列，用于做分布式消息处理逻辑
func (c *MQueueSt)InitMsgQueue(config *Config, handle MQueueNotifyHandle) {
	c.config = config
	c.handle = handle
	if c.config == nil || len(c.config.MQNameSrv) < 1 {
		return
	}
	c.mq = core.NewKafkaMQ(c.config.MQNameSrv...)
	go func() {//心跳检测，维护自身的im队列是否存活者
		for {
			stime := time.Now().Unix()
			logics.CRedis.HSet(IMMQKEY, c.config.MQueue, stime)
			log.Write(log.INFO, "检测队列topic的心跳记录到redis当中...")
			<-time.After(time.Second* HEARTINTVAL)
		}
	}()
	if c.handle != nil {//执行消费者业务逻辑
		go c.mq.ConsumerStart(IMGROUP, c.config.MQueue, func(ext *sarama.ConsumerMessage) {
			args := QueueMessageSt{}
			if err := json.Unmarshal(ext.Value, &args); err != nil {
				log.Write(log.ERROR, "queue callback json unmarshal error ", err)
				return
			}
			c.handle(config.Appid, &args) //处理消息回调业务处理  有agent代理消息发送
		})
	}
	//初始化消息的生产者处理逻辑
	c.mq.ProducerStart()
}

//判断是否已经支持了消息队列
func (c *MQueueSt) IsSupportMQ() bool {
	if c.mq != nil {
		return true
	}
	return false
}

//创建消息插入业务队列当中
func (c *MQueueSt)IntoMQueue(mqTopic string, qmsg *QueueMessageSt) {
	bstr, err := json.Marshal(qmsg)
	if err != nil {
		log.Write(log.ERROR, "插入消息队列处理失败", err)
		return
	}
	if err := c.mq.Send(mqTopic, bstr); err != nil {
		log.Write(log.ERROR, "MQ消息加入生产者{",mqTopic,"}失败", err)
	}
	log.Write(log.INFO, "MQ消息加入生产者{",mqTopic,"}OK")
}

//创建消息插入业务队列当中-广播的情况
func (c *MQueueSt)IntoAllMQueue(qmsg *QueueMessageSt) {
	bstr, err := json.Marshal(qmsg)
	if err != nil {
		log.Write(log.ERROR, "插入消息队列处理失败", err)
		return
	}
	topicMap := logics.CRedis.HGetAll(IMMQKEY).Val()
	log.Write(log.INFO, "获取到的队列:", len(topicMap))
	for mqTopic, stime := range topicMap {
		//已超过三次没有心跳记录 直接跳过 且 不能属于当前用户
		log.Write(log.INFO, "获取的队列数值情况:", mqTopic, stime, c.config.MQueue)
		vtime, _ := strconv.ParseInt(stime, 10, 64)
		if len(mqTopic) < 1 || mqTopic == c.config.MQueue || time.Now().Unix() - vtime > HEARTINTVAL*3 {
			continue
		}
		if err = c.mq.Send(mqTopic, bstr); err != nil {
			log.Write(log.ERROR, "MQALL消息加入生产者{",mqTopic,"}失败", err)
		}
		log.Write(log.INFO, "MQALL消息加入生产者{",mqTopic,"}OK")
	}
}
